package com.practice.car.cardataapp.schelduer.spark.task.sale;

import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.SaleRunner;
import lombok.Data;
import com.alibaba.fastjson.JSON;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import scala.Serializable;
import scala.Tuple2;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

public class suvtest implements SaleRunner {

    @Override
    public void run() {
        String path = Schedule.saleHdfsUrl;
        List<Car> cars = readJsonArray(path, Car.class);
        SparkConf sparkConf = new SparkConf().setAppName("wordCountApp").setMaster(Schedule.sparkMaster);
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        JavaRDD<Car> rdd=sc.parallelize(cars);
        List list=type_sales(rdd);
        sc.close();
    }
    public static <T> List<T> readJsonArray(String path, Class<T> clasz) {
        List<T> list = null;
        int available = 0;
        Path path1 = new Path(path);
        Configuration configuration = new Configuration();
        FSDataInputStream fsDataInputStream = null;
        FileSystem fileSystem = null;
        BufferedReader br = null;
        try{
            fileSystem = path1.getFileSystem(configuration);
            fsDataInputStream = fileSystem.open(path1);
            BufferedInputStream inputStream = new BufferedInputStream(fsDataInputStream);
            available = inputStream.available();
            byte[] buffer = new byte[available];
            inputStream.read(buffer);
            inputStream.close();
            list = JSON.parseArray(new String(buffer, StandardCharsets.UTF_8), clasz);
        }
        catch (IOException e) {
            e.printStackTrace();
        }

        return list;
    }

    public static List type_sales(JavaRDD<Car> rdd){
        int sum=0;
        List<Type_sales> tSaleList=new ArrayList<Type_sales>();
        JavaPairRDD rdd1=rdd.mapToPair(car -> new Tuple2<>(car.getType(),car.getSale()))
                .foldByKey(2, new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1+v2;
                    }
                });
        List list=rdd1.collect();

        for (int i = 0; i < list.size(); i++) {
            Type_sales typeSale=new Type_sales();
            String result0 =list.get(i).toString();
            String result1=result0.replace("(","");
            String result2=result1.replace(")","");
            String [] result=result2.split(",");
            typeSale.setType(result[0]);
            typeSale.setSale(Integer.parseInt(result[1]));
            tSaleList.add(i,typeSale);
            sum+=tSaleList.get(i).getSale();
        }

        for (int i = 0; i < tSaleList.size(); i++) {
            double proportion=Double.parseDouble(
                    new DecimalFormat("#.0").format((double)tSaleList.get(i).getSale()/(double)sum*100)
            );
            tSaleList.get(i).setProportion(proportion);
        }
        try {
            save(tSaleList);
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return tSaleList;
    }
    public static void save(List<Type_sales> list)throws SQLException, ClassNotFoundException{
        PreparedStatement psql;
        Connection con = Schedule.getConnection();
        psql = con.prepareStatement("DELETE FROM `suv_and_mpv_sales_ratio`");
        psql.executeUpdate();
        psql = con.prepareStatement("insert into suv_and_mpv_sales_ratio" +
                "(idSUV_and_MPV_sales_ratio,type,proportion)" +
                "values(?,?,?)");
        for (int i = 0; i <list.size() ; i++) {
            psql.setInt(1,(i+1));
            psql.setString(2,list.get(i).getType());
            psql.setDouble(3,list.get(i).getProportion());
            psql.executeUpdate();
        }
        con.close();
    }

    @Data
    static class Car implements Serializable {
        String name;
        String brand;
        String type;
        double price;
        String date;
        int sale;
        String national;

        Car() {
            name = "";
            brand = "";
            type = "";
            price = 0;
            date = "";
            sale = 0;
            national = "";
        }
    }

    @Data
    static class Type_sales{
        String type;
        int sale;
        double proportion;

        Type_sales(){
            type="";
            sale=0;
            proportion=0;
        }
    }

}

